package hadoop.hdfs.demo;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**
 * @author qiss
 * @create 2018-02-12 18:06
 **/
public class LogEnhanOutputFormat extends FileOutputFormat<Text,NullWritable> {
    public RecordWriter<Text, NullWritable> getRecordWriter(TaskAttemptContext context) throws IOException, InterruptedException {
        FileSystem fs = FileSystem.get(context.getConfiguration());
        Path enhancePath = new Path("hdfs://hadoop-namenode1:9000/enhancePath");
        Path toCrawPath = new Path("hdfs://hadoop-namenode1:9000/toCrawPath");
        FSDataOutputStream enhanceOut = fs.create(enhancePath);
        FSDataOutputStream toCrawlOut = fs.create(toCrawPath);
        return new MyRecordWrite(enhanceOut,toCrawlOut);


    }

    private class MyRecordWrite extends RecordWriter<Text, NullWritable> {
        FSDataOutputStream enhanceOut = null;
        FSDataOutputStream toCrawlOut = null;
        public MyRecordWrite(FSDataOutputStream enhanceOut, FSDataOutputStream toCrawlOut) {
            this.enhanceOut = enhanceOut;
            this.toCrawlOut = toCrawlOut;
        }

        public void write(Text text, NullWritable nullWritable) throws IOException, InterruptedException {
            if(text.toString().contains("tocrawl")){
                toCrawlOut.write(text.toString().getBytes());
            }else{
                enhanceOut.write(text.toString().getBytes());
            }
        }

        public void close(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
            if(toCrawlOut!=null){
                toCrawlOut.close();
            }
            if(enhanceOut!=null){
                enhanceOut.close();
            }

        }
    }
}
